refactor(lifecycle): one module owns all job/run status transitions by seandavi · Pull Request #166 · seandavi/nextflow_telemetry

seandavi · 2026-07-10T21:18:01Z

What

Extracts every inline update(...).values(status=...) write on jobs / workflow_runs into a new services/lifecycle.py — a module of free functions taking conn: AsyncConnection (caller owns the transaction). Introduces JobStatus / RunStatus StrEnums as the status vocabulary; DB columns stay String (no migration).

This is roadmap "Architecture deepening" candidate #3, designed in a grilling session before build.

Design (agreed)

Connection seam — methods take conn, never open engine.begin(); composes with the existing atomic multi-step call sites.
Tolerant / guarded / idempotent — legal source-states live in the WHERE clause; an event matching nothing is a silent no-op, never raises (absorbs out-of-order/duplicate weblog events).
Threshold-free (caller passes now/cutoff) and HTTP-agnostic (returns plain values; routers map to 404).
Functions: claim, mark_submitted, mark_running, complete_sample, close_run, sweep_incomplete (moved verbatim from reconcile.sweep_run_incomplete), requeue_expired, requeue_dead_letter, reset_jobs_to_pending.
Carve-outs deliberately NOT owned here (documented in the module docstring): job birth (ReconcileService.reconcile_jobs) and retire-time pending-job purge (WorkflowService) — population ops that live with their policy.

Reroutes telemetry.py, dispatch.py, admin.py, runs.py; deletes the moved sweep from reconcile.py (no re-export shim); folds the duplicate claimed→submitted write in runs.py _apply_summary_update into mark_submitted.

Behaviour

Behaviour-preserving in every tested path. One deliberate divergence, documented at the call site: close_run is now idempotent, so a late completed weblog event no longer clobbers a run the heartbeat watchdog already marked failed.

Tests

New tests/test_lifecycle.py (13 tests) drives transitions through one connection at the module seam — happy path, tolerant no-ops, sweep retry-vs-DLQ branches, requeue_expired no-retry-penalty, close_run idempotency.

just typecheck: clean (59 files)
just test: 170 passed (integration + e2e on real testcontainers Postgres)

Reviewed

Built by a separate agent, then reviewed on two axes (Standards + Spec) — no hard violations; the terminal-status duplication that surfaced was fixed (public RUN_TERMINAL_STATUSES).

🤖 Generated with Claude Code

…review Adds a 12-row deepening table (strength / status / seam / cross-ref), a "solid ground — don't touch" list, and a start-here ordering. Candidate 1 (membership seam) marked done (#164/#165). Durable record of the review so it survives context clears. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Extracts every inline `update(...).values(status=...)` write on jobs_tbl / workflow_runs_tbl into a new services/lifecycle.py — a module of free functions taking `conn: AsyncConnection` (caller owns the transaction). Introduces JobStatus / RunStatus StrEnums as the status vocabulary; DB columns stay String (no migration). Design (grilled + agreed, roadmap "Architecture deepening" candidate #3): - Connection seam: methods take conn, never open engine.begin(). - Tolerant/guarded/idempotent: legal source-states live in the WHERE clause; an event matching nothing is a silent no-op, never raises — absorbs out-of-order/duplicate weblog events. - Threshold-free (caller passes now/cutoff) and HTTP-agnostic (returns plain values; routers map to 404). - Functions: claim, mark_submitted, mark_running, complete_sample, close_run, sweep_incomplete (moved verbatim from reconcile.sweep_run_incomplete), requeue_expired, requeue_dead_letter, reset_jobs_to_pending. - Carve-outs deliberately NOT owned here (documented in the module docstring): job birth (ReconcileService.reconcile_jobs) and retire-time pending-job purge (WorkflowService) — population ops that live with their policy. Reroutes telemetry.py, dispatch.py, admin.py, runs.py; deletes the moved sweep from reconcile.py (no re-export shim). Folds the duplicate claimed->submitted write in runs.py `_apply_summary_update` into mark_submitted. Behaviour-preserving in every tested path (170 tests green, incl. integration + e2e). One deliberate divergence, documented at the call site: close_run is idempotent, so a late `completed` weblog event no longer clobbers a run the heartbeat watchdog already marked `failed`. New tests/test_lifecycle.py (13 tests) drives the transitions through one connection at the module seam — happy path, tolerant no-ops, sweep retry-vs-DLQ branches, requeue_expired no-retry-penalty, close_run idempotency. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Copilot

Pull request overview

This PR centralizes all jobs and workflow_runs status transitions into a new services/lifecycle.py module (with JobStatus / RunStatus StrEnum vocab), and rewires routers/services to call those transition helpers instead of doing inline update(...).values(status=...) writes. It also adds a dedicated lifecycle test suite and updates the roadmap with the “Architecture deepening” design review notes.

Changes:

Added services/lifecycle.py to own job/run status transitions and moved “sweep incomplete” logic into it.
Updated dispatch/admin/telemetry/runs codepaths to call lifecycle functions (and removed the old sweep_run_incomplete from services/reconcile.py).
Added tests/test_lifecycle.py to exercise lifecycle transitions directly against Postgres.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
tests/test_lifecycle.py	New contract-style tests for lifecycle transitions via a single AsyncConnection seam.
src/nextflow_telemetry/services/telemetry.py	Replaced inline status updates with lifecycle calls; documents close-run idempotency behavior change.
src/nextflow_telemetry/services/reconcile.py	Removes sweep helper; clarifies reconcile now only handles job “birth”.
src/nextflow_telemetry/services/lifecycle.py	New single-owner module for job/run status writes and related transition logic.
src/nextflow_telemetry/routers/runs.py	Uses lifecycle `mark_submitted` for wrapper-started fallback transition.
src/nextflow_telemetry/routers/dispatch.py	Uses lifecycle for claim/submitted/expiry requeue transitions.
src/nextflow_telemetry/routers/admin.py	Uses lifecycle for close/sweep/reset/requeue dead letter; uses shared terminal status set.
docs/roadmap.md	Adds/updates “Architecture deepening” section and updates last-updated date.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+    await conn.execute(
+        update(workflow_runs_tbl)
+        .where(workflow_runs_tbl.c.run_name == run_name)
+        .values(run_id=run_id, status=RunStatus.running, started_at=now)
+    )


+    result = await conn.execute(
+        update(jobs_tbl)
+        .where(
+            jobs_tbl.c.run_name == run_name,
+            jobs_tbl.c.sample_id == sample_id,
+        )
+        .values(status=JobStatus.completed, completed_at=now)
+    )


+async def mark_submitted(
+    conn: AsyncConnection,
+    run_name: str,
+    executor_job_id: str | None,
+) -> bool:
+    """Transition a run + its jobs from `claimed` to `submitted`.
+
+    Returns True iff a workflow_runs row matched (was in `claimed`) and was
+    updated; the caller maps False to a 404. Mirrors routers/dispatch.py
+    `report_submitted`.
+    """
+    now = datetime.now(timezone.utc)
+    result = await conn.execute(
+        update(workflow_runs_tbl)
+        .where(
+            workflow_runs_tbl.c.run_name == run_name,
+            workflow_runs_tbl.c.status == RunStatus.claimed,
+        )
+        .values(
+            status=RunStatus.submitted,
+            submitted_at=now,
+            executor_job_id=executor_job_id,
+        )
+        .returning(workflow_runs_tbl.c.run_name)


…eview) Addresses the four Copilot review comments on #166 — three found the module docstring claims "guarded/idempotent" but some functions didn't fully honor it. Extends the close_run no-clobber precedent uniformly: - mark_running: guard the workflow_runs update against terminal states so a late/duplicate `started` can't resurrect a completed/failed/expired run. - complete_sample: guard against terminal job states so a late MARK_COMPLETE can't flip a `failed` job to `completed` (which would leave failure fields populated on a completed row). - close_run: SELECT ... FOR UPDATE so a watchdog(failed) vs telemetry(completed) race in separate transactions can't lost-update each other. - mark_submitted: take `now: datetime` as a parameter instead of reading the clock internally — restores the agreed threshold-free contract; callers (dispatch.report_submitted, runs._apply_summary_update) pass it. Adds two tests (mark_running no-clobber-terminal, complete_sample no failed->completed). typecheck clean; full suite 172 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Copilot

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

+    await conn.execute(
+        update(jobs_tbl)
+        .where(jobs_tbl.c.id.in_(job_ids))
+        .values(run_name=run_name, status=JobStatus.claimed)
+    )


…opilot round 2) Two further guards from the re-review, both idempotency-consistency: - claim: guard the jobs update on `status == pending` so a bad/future caller can't clobber an already-claimed/running job's status+run_name. The sole caller (pick-then-lock) already selects pending ids, so this is a no-op in the real flow — defense for the imminent DispatchService refactor (#4). - mark_running: restrict the run update to its legal predecessors (claimed/submitted) instead of just "non-terminal", so a duplicate/late `started` no longer rewrites started_at/run_id on an already-running run (which skewed queue-wait timing). Matches the jobs-side guard. Adds two tests (duplicate-started timing no-op; claim skips non-pending job). Full suite 174 passed; typecheck clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

seandavi · 2026-07-10T21:44:34Z

Thanks @copilot — addressed all findings across two rounds:

Round 1

mark_submitted now takes now: datetime (restores the threshold-free contract).
complete_sample guarded against terminal job states (no failed→completed flip).
close_run SELECT ... FOR UPDATE (prevents the watchdog/telemetry lost-update race).
mark_running guarded against terminal runs.

Round 2

claim guards the jobs update on status == pending (no clobber of already-claimed jobs).
mark_running run update restricted to legal predecessors (claimed/submitted) so a duplicate/late started no longer rewrites started_at/run_id.

Added 4 tests for these guards. Full suite: 174 passed, typecheck clean.

) * docs(roadmap): mark deepening candidate #3 (job-lifecycle) done (#166) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(dispatch): extract DispatchService from the HTTP handlers Pulls the dispatch orchestration out of routers/dispatch.py into a new services/dispatch.py (@DataClass DispatchService(engine)), following the service-injected-router convention (services/sample.py + routers/samples.py) — the one clean wiring the design review praised. Deepening candidate #4. - claim_batch(): the two-phase pick-then-lock (SELECT ... FOR UPDATE SKIP LOCKED, issue #74), run_name minting, lifecycle.claim(), and sample-metadata fetch — moved verbatim. Returns a plain ClaimedBatch/ClaimedJob dataclass or None; the service is HTTP/Pydantic-agnostic (no FastAPI imports). - report_submitted() -> bool, requeue_expired() -> int; CLAIM_TTL_MINUTES and the uuid7 version-guard move here too. - The state writes still go through services/lifecycle.py (the claim seam from #166); this owns only the orchestration around them. Router drops to a thin call-and-map: parse request -> service -> map None to 204 / False to 404. Pydantic models stay inline (repo convention); all OpenAPI strings unchanged. Behaviour-preserving. New tests/test_dispatch_service.py drives the service directly at the seam (claim-then-empty, submitted true/false, requeue-expired). typecheck clean; full suite 177 passed (existing dispatch integration/e2e unchanged). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(dispatch): seed claimed job with run_name in one transaction (Copilot) Copilot review nit on #167: the requeue-expired test seeded a claimed job with run_name=NULL then set it in a second transaction. Seed it in one insert via a new run_name param on the _seed_job helper. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

seandavi and others added 2 commits July 10, 2026 13:20

seandavi requested a review from Copilot July 10, 2026 21:18

Copilot started reviewing on behalf of seandavi July 10, 2026 21:19 View session

Copilot AI reviewed Jul 10, 2026

View reviewed changes

seandavi requested a review from Copilot July 10, 2026 21:32

Copilot started reviewing on behalf of seandavi July 10, 2026 21:32 View session

Copilot AI reviewed Jul 10, 2026

View reviewed changes

Comment thread src/nextflow_telemetry/services/lifecycle.py

Comment on lines +125 to +129

await conn.execute(

update(jobs_tbl)

.where(jobs_tbl.c.id.in_(job_ids))

.values(run_name=run_name, status=JobStatus.claimed)

)

Comment thread src/nextflow_telemetry/services/lifecycle.py

seandavi merged commit db6fbe0 into main Jul 10, 2026
1 check failed

seandavi deleted the refactor/job-lifecycle-module branch July 10, 2026 21:44

Copilot stopped work on behalf of seandavi due to an error July 10, 2026 21:45
The session was cancelled by the user.

seandavi mentioned this pull request Jul 10, 2026

refactor(dispatch): extract DispatchService from the HTTP handlers #167

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(lifecycle): one module owns all job/run status transitions#166

refactor(lifecycle): one module owns all job/run status transitions#166
seandavi merged 4 commits into
mainfrom
refactor/job-lifecycle-module

seandavi commented Jul 10, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

seandavi commented Jul 10, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

seandavi commented Jul 10, 2026

What

Design (agreed)

Behaviour

Tests

Reviewed

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

seandavi commented Jul 10, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants